package ar.edu.itba.infovis.TPF.twitter;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;

import java.net.URLEncoder;

import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.UnrecoverableKeyException;
import java.security.cert.CertificateException;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.PropertyResourceBundle;
import java.util.ResourceBundle;
import java.util.Set;
import java.util.SortedSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.servlet.http.HttpServletResponse;

import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponse;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.AuthCache;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.protocol.ClientContext;
import org.apache.http.impl.auth.BasicScheme;
import org.apache.http.impl.client.BasicAuthCache;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.protocol.BasicHttpContext;

import org.codehaus.jackson.map.ObjectMapper;

import ar.edu.itba.infovis.TPF.MainServlet;

import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;

public class TwitterStream {

	private static final int MIN_KEYWORDS = 1;
	private static final int MAX_KEYWORDS = MainServlet.MAXSUBJECTS;
	
	private static String username = "StreamViz";
	private static String password = "itba2012";
	
	private static final String TWITTER_STREAM_FILTER_API_HOST = "stream.twitter.com";
	private static final String TWITTER_STREAM_FILTER_API_URL = "/1/statuses/filter.json";
	private static final int TWITTER_STREAM_FILTER_API_PORT = 443;
	private static final String TWITTER_STREAM_FILTER_API_PROTOCOL = "https";

	private static final String HTTP_QUERY_SEPARATOR = "?";
	private static final String TWITTER_STREAM_FILTER_API_TRACK_KEYWORD = "track=";
	private static final String TWITTER_KEYWORD_SEPARATOR = ",";
	
	private static TwitterStream instance;
	
	public static TwitterStream getInstance(){
		if( instance == null ){
			synchronized(TwitterStream.class){
				if( instance == null) {
					instance = new TwitterStream();
				}
			}
		}
		return instance;
	}
	
	private static final HttpHost twitterHost = new HttpHost(TWITTER_STREAM_FILTER_API_HOST, TWITTER_STREAM_FILTER_API_PORT, TWITTER_STREAM_FILTER_API_PROTOCOL);
	
	private ConcurrentHashMap<String, SortedSet<Tweet>> streams;
	private ExecutorService executorService;
	private DefaultHttpClient httpclient;
	
	private TwitterStream(){
		this.streams = new ConcurrentHashMap<String, SortedSet<Tweet>>();
		this.executorService = Executors.newCachedThreadPool();
		this.httpclient = new DefaultHttpClient();
		
		ResourceBundle bundle; 
        try { 
             bundle = new PropertyResourceBundle(new FileInputStream("conf.properties")); 
        } catch (IOException e) { 
             System.err.println("ERROR: Could not find application configuration (conf.properties)"); 
             return; 
        } 
        username = bundle.containsKey("username") ? bundle.getString("username") : username;
        password = bundle.containsKey("password") ? bundle.getString("password") : password;
	}
	
	private String buildURL(List<String> keywords) throws UnsupportedEncodingException{
		Preconditions.checkNotNull(keywords, "List of keywords can't be null");
		Preconditions.checkArgument(!keywords.isEmpty(), "List of keywords can't be empty");
		Preconditions.checkArgument(keywords.size() >= MIN_KEYWORDS, "List of keywords must have at least " + MIN_KEYWORDS + " keywords");
		Preconditions.checkArgument(keywords.size() <= MAX_KEYWORDS, "List of keywords must have at most " + MAX_KEYWORDS +  " keywords");
		
		StringBuffer buf = new StringBuffer(TWITTER_STREAM_FILTER_API_URL);
		buf.append(HTTP_QUERY_SEPARATOR);
		buf.append(TWITTER_STREAM_FILTER_API_TRACK_KEYWORD);
		
		Iterator<String> it = keywords.iterator();
		while(it.hasNext()){
			buf.append(URLEncoder.encode(it.next(), "UTF-8"));
			if( it.hasNext()) {
				buf.append(TWITTER_KEYWORD_SEPARATOR);
			}
		}
		
		return buf.toString();
	}
	
	public synchronized void addTweets(String keyword, List<Tweet> tweets){
		SortedSet<Tweet> tweetSet = streams.get(keyword);
		if( tweetSet != null ){
			tweetSet.addAll(tweets);
		}
	}
	
	public synchronized List<String> getKeywords() {
		Set<String> keys = streams.keySet();
		ArrayList<String> list = new ArrayList<String>();
		list.addAll(keys);
		return list;
	}
	
	public synchronized SortedSet<Tweet> getTweets(String keyword){
		return streams.get(keyword);
	}
	
	public synchronized void stream(final List<String> keywords, boolean deleteHistory){
		if( !executorService.isShutdown() ){
			this.executorService.shutdownNow();
			getHttpClient().getConnectionManager().shutdown();
			this.executorService = Executors.newCachedThreadPool();
		}
		
		if( deleteHistory ){
			streams.clear();
		}
		
		executorService.execute(new Runnable() {
			@Override
			public void run() {
				try {
					streamThread(keywords);
				} catch (Exception e) {
					e.printStackTrace();
				} 
			}
		});
		
	}
	
	private synchronized DefaultHttpClient getHttpClient(){
		return httpclient;
	}
	
	private synchronized void setHttpClient(DefaultHttpClient httpclient){
		this.httpclient = httpclient;
	}
	
	private void streamThread(List<String> keywords) throws ClientProtocolException, IOException, NoSuchAlgorithmException, CertificateException, KeyStoreException, KeyManagementException, UnrecoverableKeyException {

		for(String keyword: keywords){
			SortedSet<Tweet> tweets = streams.get(keyword);

			if( tweets == null ) {
				tweets = new ConcurrentSkipListSet<Tweet>();
				streams.put(keyword, tweets);
			}
		}
		
		/*ResourceBundle bundle; 
        try { 
             bundle = new PropertyResourceBundle(new FileInputStream("conf.properties")); 
        } catch (IOException e) { 
             System.err.println("ERROR: Could not find application configuration"); 
             return; 
        } 
        String useHistory = bundle.containsKey("useHistory") ? bundle.getString("useHistory") : null;
		if( "true".equals(useHistory) ){
			TODO: Save reference to this?
			new TwitterSearch().search(keywords);
		}*/
		
		
		String url = this.buildURL(keywords);
		
        setHttpClient(new DefaultHttpClient());
        httpclient.getCredentialsProvider().setCredentials(
                new AuthScope(twitterHost.getHostName(), twitterHost.getPort()),
                new UsernamePasswordCredentials(username, password));

        AuthCache authCache = new BasicAuthCache();
        authCache.put(twitterHost, new BasicScheme());
        BasicHttpContext localcontext = new BasicHttpContext();
        localcontext.setAttribute(ClientContext.AUTH_CACHE, authCache);

        System.out.println(url);
        HttpGet httpGet = new HttpGet(url);
        
        System.out.println("executing request: " + httpGet.getRequestLine());
        System.out.println("to target: " + twitterHost);

        HttpResponse response = httpclient.execute(twitterHost, httpGet, localcontext);
        HttpEntity entity = response.getEntity();
        
        if( response.getStatusLine().getStatusCode() != HttpServletResponse.SC_OK ){
        	System.err.println("Twitter API request failed. Response code: " + response.getStatusLine().getStatusCode());
        	return;
        }

        System.out.println(response.getStatusLine());
        if (entity != null) {
            InputStream is = entity.getContent();
            BufferedReader reader = new BufferedReader(new InputStreamReader(is, Charsets.UTF_8));
            ObjectMapper mapper = new ObjectMapper();
            
            long tweetCount = 0;
            long millis = System.currentTimeMillis();
            
            while(true){
            	
            	if( (System.currentTimeMillis() - millis) >= 1000){
            		System.out.println("Frequency: " + tweetCount + "/sec");
            		tweetCount = 0;
            		millis = System.currentTimeMillis();
            	} else {
            		tweetCount++;
            	}
            	
            	String rawJSON = null;
            	try{
            		rawJSON = reader.readLine();
            	} catch (IllegalStateException e){
            		System.err.println("Stream closed");
            		return;
            	}
            	if( rawJSON == null || rawJSON.isEmpty() ){
            		System.err.println("Retrieved JSON was null or empty. This is probably a stream too big to handle");
            		continue;
            	}
            	Tweet t = mapper.readValue(rawJSON, Tweet.class);

            	if( t.getText() == null ){
            		System.err.println("Tweet text was null. This is probably a stream too big to handle");
            		continue;
            	}
            	
//           	System.out.println(t.getText());
//            	System.out.println("@" + t.getUser().getScreen_name());
//            	System.out.println(t.getUser().getFollowers_count());
//            	System.out.println(t.getCreated_at().toString(DateTimeFormat.forPattern("E M d H:m:s Z Y")));
//            	System.out.println("------");
            	
            	String tweetText = t.getText().toLowerCase();
            	
            	for(String k: keywords){
            		if( tweetText.contains(k) ){
            			if(tweetText.contains("RT")) {
            				t.setRetweet_count(1);
            			}
            			streams.get(k).add(t);
            		}
            	}
            	
            }
        }
//        EntityUtils.consume(entity);

        getHttpClient().getConnectionManager().shutdown();
	}
	
}
